package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.BaseDBTableProcessFunction;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * @author Felix
 * @date 2023/1/3
 * 事实表动态分流
 * dwd事实表处理划分如下几个数据域
 *      流量域
 *          错误日志、启动日志、页面日志、曝光日志、动作日志
 *      交易域
 *          加购、下单、取消订单、支付成功、退单、退款成功
 *      工具域
 *          优惠券领用、优惠券下单、优惠券支付
 *      互动域
 *          收藏、评论
 *      用户域
 *          用户注册
 * 简单事实表处理动态分流实现
 *      -基本环境准备
 *      -检查点相关的设置
 *      -从topic_db主题中读取数据
 *      -简单ETL并对流中数据类型进行转换  jsonStr->jsonObj
 *      ~~~~~~~~~~~~~~~~~~~~~~~~读取主流业务数据~~~~~~~~~~~~~~~~~~~~~~~~~~
 *      -使用FlinkCDC读取配置表数据---配置流
 *      -将配置流进行广播---广播流
 *      ~~~~~~~~~~~~~~~~~~~~~~~读取配置数据~~~~~~~~~~~~~~~~~~~~~~~~~~
 *      -将主流和广播流进行关联---connect
 *      -对关联之后的数据进行处理---process
 *          class BaseDBTableProcessFunction extends BroadcastProcessFunction{
 *              open
 *                  配置信息预加载
 *              processElement-处理主流数据
 *                  获取当前处理业务数据表名以及操作类型
 *                  将表名和操作类型拼接为key
 *                  根据key到广播状态中以及自己维护的Map集合中去拿对应的配置信息
 *                  如果配置信息不为空，说明当前处理的数据是事实数据，向下游传递
 *                      过滤掉不需要传递的属性
 *                      补充输出目的地
 *
 *              processBroadcastElement-处理广播数据
 *                  if(op=="d"){
 *                      将配置信息从广播状态中删除
 *                      将配置信息从自己维护的map中删除
 *                  }else{
 *                      将配置信息放到广播状态  map<k-表名:操作类型,v-TableProcess对象>
 *                      将配置信息放到自己维护的map中
 *                  }
 *          }
 * 将流中的数据发送到kafka的不同主题中
 */
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);

        env.enableCheckpointing(5000L);
        /*
        //TODO 2.检查点相关的设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消后  检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop202:8020/gmall/ck");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */
        //TODO 3.从kafka的topic_db主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "topic_db";
        String groupId = "base_db_group";

        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //TODO 4.简单的ETL以及类型转换       jsonStr->jsonObj----主流业务数据
        SingleOutputStreamOperator<JSONObject> filterDS = kafkaStrDS.process(
            new ProcessFunction<String, JSONObject>() {
                @Override
                public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                    try {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        out.collect(jsonObj);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        );
        // filterDS.print(">>>");

        //TODO 5.使用FlinkCDC读取配置表数据--配置流
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("hadoop202")
            .port(3306)
            .databaseList("gmall0725_config") // set captured database
            .tableList("gmall0725_config.table_process") // set captured table
            .username("root")
            .password("123456")
            .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
            .startupOptions(StartupOptions.initial())
            .build();

        // 6.2 封装为流
        DataStreamSource<String> mysqlDS = env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "base-db-mysql-source");

        //TODO 6.将读取到的配置流进行广播--广播流
        MapStateDescriptor<String, TableProcess> mapStateDescriptor
            = new MapStateDescriptor<String, TableProcess>("mapStateDescriptor",String.class,TableProcess.class);
        BroadcastStream<String> broadcastDS = mysqlDS.broadcast(mapStateDescriptor);

        //TODO 7.将主流和广播流进行关联
        BroadcastConnectedStream<JSONObject, String> connectDS = filterDS.connect(broadcastDS);

        //TODO 8.对关联之后的数据进行处理（过滤出我们要动态分流的事实表）
        SingleOutputStreamOperator<JSONObject> realDS = connectDS.process(
            new BaseDBTableProcessFunction(mapStateDescriptor)
        );
        //TODO 9.将不同事实表的数据  写到kafka的不同主题中
        realDS.print(">>>");

        realDS
            .addSink(
               MyKafkaUtil.getKafkaProducerBySchema(new KafkaSerializationSchema<JSONObject>() {
                   @Override
                   public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj, @Nullable Long timestamp) {
                       String kafkaTopic = jsonObj.getString("sink_table");
                       jsonObj.remove("sink_table");
                       return new ProducerRecord<byte[], byte[]>(kafkaTopic,jsonObj.toJSONString().getBytes());
                   }
               })
            );

        env.execute();
    }
}
